package io.mqttpush.mqttclient.conn;

import io.mqttpush.common.events.MsgEvent;
import io.mqttpush.mqttclient.handle.PingRunnable;
import io.mqttpush.mqttclient.handle.PubHandle;
import io.mqttpush.mqttclient.handle.SubHandle;
import io.mqttpush.mqttclient.service.DefaultMessageListener;
import io.mqttpush.mqttclient.service.MessageListener;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

public class Connetor {

    Logger logger = LoggerFactory.getLogger(Connetor.class);
    final EventLoopGroup eventLoopGroup;
    final Bootstrap bootstrap;

    final ConnectorOption connectorOption;
    final DefaultOptionsProvider apiService;
    final MessageListener messageListener;

    final ConnectionHandle connectionHandle;


    private Consumer afterConnect;


    CancelbleExecutorService executorService = new CancelbleExecutorService(2);


    public Connetor(ConnectorOption connectorOption) {
        this(connectorOption, new DefaultMessageListener());
    }

    public Connetor(ConnectorOption connectorOption, MessageListener messageListener) {
        super();

        this.connectorOption = connectorOption;
        this.messageListener = messageListener;


        eventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
        apiService = DefaultOptionsProvider.instance();


        connectionHandle = new ConnectionHandle(Connetor.this, apiService,
                this.connectorOption.getDeviceId(),
                this.connectorOption.getUsername(),
                this.connectorOption.getPassword());

        bootstrap = new Bootstrap();


        init();
    }

    /**
     * 初始化
     */
    public void init() {


        if (connectorOption.getConnectTimeOut() > 0) {
            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectorOption.getConnectTimeOut());
        }

        bootstrap.option(ChannelOption.TCP_NODELAY, true);


        bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();

                        p.addLast(MqttEncoder.INSTANCE, new MqttDecoder(),
                                connectionHandle,
                                new PubHandle(messageListener), new SubHandle());
                    }
                });


        connection();
    }

    /**
     * 连接
     *
     * @return
     */
    protected ChannelFuture connection() {

        String host = connectorOption.getHost();
        Integer port = connectorOption.getPort();


        ChannelFuture channelFuture = bootstrap.connect(host, port);


        channelFuture.addListener((ChannelFuture future) -> {
            whenConnectFutureResult(future, host, port);
        });

        return channelFuture;
    }


    private void whenConnectFutureResult(ChannelFuture future, String host, Integer port) {


        AtomicBoolean hasConnect = Status.hasConnect;

        if (future.isSuccess()) {

            hasConnect.set(true);

            PingRunnable pingRunnable = new PingRunnable(future.channel(),
                    Connetor.this,
                    connectorOption.getPingTimeOut(),
                    executorService);

            pingRunnable.updatehasResp(true);
            connectionHandle.setPingRunnable(pingRunnable);
            executorService.schedule(pingRunnable, connectorOption.getPingTimeOut(), TimeUnit.SECONDS);


            if (logger.isDebugEnabled()) {
                logger.debug("connect the server successful!{},{}", host, port);
            }


            if (afterConnect != null) {
                afterConnect.accept(Connetor.this);
            }




        } else {
            /**
             * 连接失败，重新连接
             */
            if (hasConnect.get()) {
                logger.info(connectorOption.getDeviceId() + "已经连接,无需重复连接");
                return;
            }
            executorService.schedule(() -> {
                reconnection(future.channel());
            }, connectorOption.getPingTimeOut(), TimeUnit.SECONDS);
            logger.warn("连接失败");
        }
    }

    /**
     * 重连接
     *
     * @return
     */
    public void reconnection(Channel channel) {


        logger.warn("开始重连");
        AtomicBoolean hasConnect = Status.hasConnect;
        /**
         * 清理调调度线程池
         */
        executorService.reset();
        hasConnect.set(false);

        if (channel != null && channel.isActive()) {
            channel.close();

        }


        connection();
    }


    /**
     * 桥接这三个方法，是外部API使用一个类就可以操作
     *
     * @param topname
     * @param bs
     * @param qoS
     * @return
     */
    public Future<?> pubMsg(String topname, byte[] bs, MqttQoS qoS) {

        return apiService.pubMsg(topname, bs, qoS);
    }


    /**
     * @param topname  订阅的主题
     * @param qoS      订阅的服务质量
     * @param consumer 收到消息的回调
     * @return
     */
    public Future<?> subscribeMessage(String topname, MqttQoS qoS, Consumer<ByteBuf> consumer) {


        messageListener.addMessageProcesser(topname, consumer);

        UnLoginTask.topics.add(new UnLoginTask.TopicAndQos(topname,qoS));
        return apiService.subscribe(topname, qoS);
    }

    /**
     * @param topname  订阅的主题
     * @param qoS      订阅的服务质量
     * @param consumer 收到消息的回调
     * @return
     */
    public Future<?> subscribeEvent(String topname, MqttQoS qoS, Consumer<MsgEvent> consumer) {


        messageListener.addEventProcesser(topname, consumer);
        UnLoginTask.topics.add(new UnLoginTask.TopicAndQos(topname,qoS));
        return apiService.subscribe(topname, qoS);
    }


    public <T> Connetor afterConnect(Consumer<T> afterConnect) {
        this.afterConnect = afterConnect;
        return this;
    }

    public Future<?> unsub(String topname) {
        return apiService.unsub(topname);
    }


}
